package org.lboot.rabbit.client;

import cn.hutool.core.lang.Validator;
import cn.hutool.core.util.HashUtil;
import cn.hutool.json.JSONUtil;
import com.rabbitmq.client.*;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.lboot.rabbit.config.RocketRabbitProps;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Data
@Component
@AllArgsConstructor
public class RocketRabbitClient {
    @Resource
    RocketRabbitProps props;

    RabbitTemplate rabbitTemplate;

    // 连接池
    private static final ConcurrentHashMap<String, Connection> clients = new ConcurrentHashMap<>();


    @SneakyThrows
    private String clientHash(String username, String password){

        String calcStr = username + password;
        return String.valueOf(HashUtil.fnvHash(calcStr));

    }

    /**
     * 使用系统默认配置
     * @return 默认连接
     */
    @SneakyThrows
    public Connection getConnection(){
        String clientKey = clientHash(props.getUsername(), props.getPassword());
        Connection connection = clients.get(clientKey);
        if (Validator.isNotEmpty(connection)){
            return connection;
        }
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(props.getHost());
        factory.setUsername(props.getUsername());
        factory.setPassword(props.getPassword());
        // 创建一个connection
        connection = factory.newConnection();
        //存入hash
        clients.put(clientKey,connection);
        return connection;
    }

    @SneakyThrows
    public Connection getConnection(String username, String password){
        String clientKey = clientHash(username, password);
        Connection connection = clients.get(clientKey);
        if (Validator.isNotEmpty(connection)){
            return connection;
        }
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(props.getHost());
        factory.setUsername(username);
        factory.setPassword(password);
        // 创建一个connection
        connection = factory.newConnection();
        //存入hash
        clients.put(clientKey,connection);
        return connection;
    }


    @SneakyThrows
    public Channel getChannel(){
        Connection connection = getConnection();
        return connection.createChannel();
    }


    @SneakyThrows
    public Channel getChannel(String username, String password){
        Connection connection = getConnection(username, password);
        return connection.createChannel();
    }



    /**
     * AMQP 协议发送信息
     * @param <T>
     */
    @SneakyThrows
    public <T> T publishMessage(String queueName, T t){
        rabbitTemplate.convertAndSend("",queueName, t);
        return t;
    }

    /**
     * 使用 Template 发送消息
     * @param queueName
     * @param t
     * @return
     * @param <T>
     */
    @SneakyThrows
    public <T> T templatePublishMessage(String queueName, T t){
        /*
         * 1. 交换器
         * 2. 队列
         * 3. 数据
         */
        rabbitTemplate.convertAndSend("",queueName, t);
        return t;
    }

    /**
     * 消息订阅
     * @return
     * @param <T>
     */
    @SneakyThrows
    public <T> T subscribeMessage(String queueName, DeliverCallback deliver, CancelCallback callback){
        Channel channel = getChannel();

        channel.basicConsume(queueName, true, deliver, callback);
        // 消费消息的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("成功消费消息,内容为:" + new String(message.getBody()));
        };
        return null;
    }


}
